Check if these issues are valid — if so, understand the root cause of each and fix them.


<file name="backend/app/services/event_bus.py">

<violation number="1" location="backend/app/services/event_bus.py:138">
P2: Synchronous handlers run directly on the event loop, which can block async processing. Offload sync handlers to a thread executor to avoid stalling Kafka message handling.</violation>
</file>

<file name="backend/workers/run_pod_monitor.py">

<violation number="1" location="backend/workers/run_pod_monitor.py:41">
P2: Ensure cleanup runs even when the consumer loop is cancelled or errors. Wrap the loop in a try/finally so container.close executes on shutdown.</violation>
</file>

<file name="backend/app/db/repositories/resource_repository.py">

<violation number="1" location="backend/app/db/repositories/resource_repository.py:145">
P1: Validate resource requests are non-negative before allocation. Negative values would increase the available pool, allowing callers to mint resources.</violation>
</file>

<file name="backend/workers/run_coordinator.py">

<violation number="1" location="backend/workers/run_coordinator.py:42">
P2: Ensure container cleanup runs even if the consumer loop raises by wrapping the loop in a try/finally and moving `container.close()` into the finally block.</violation>
</file>

<file name="backend/workers/run_k8s_worker.py">

<violation number="1" location="backend/workers/run_k8s_worker.py:41">
P2: Wrap the consume loop in `try/finally` so the DI container is closed even when message handling raises, otherwise resources remain open and the worker can leak connections on failures.</violation>
</file>

<file name="backend/app/services/pod_monitor/monitor.py">

<violation number="1" location="backend/app/services/pod_monitor/monitor.py:187">
P2: Pod events are no longer persisted to the EventRepository before producing to Kafka. This removes the audit copy previously stored via KafkaEventService and can break consumers that rely on stored events.</violation>
</file>

<file name="backend/workers/run_event_replay.py">

<violation number="1" location="backend/workers/run_event_replay.py:33">
P2: On shutdown (cancellation), `await asyncio.Event().wait()` raises `CancelledError` and the container is never closed, leaking resources. Wrap the wait in a try/finally so `container.close()` always runs.</violation>
</file>

<file name="backend/app/events/core/consumer.py">

<violation number="1" location="backend/app/events/core/consumer.py:47">
P2: Guard against missing Kafka headers. This line will crash on messages produced without headers (producer sends `headers=None`). Use a fallback for `msg.headers` and decode only when a value is present.</violation>
</file>

<file name="backend/app/db/repositories/execution_queue_repository.py">

<violation number="1" location="backend/app/db/repositories/execution_queue_repository.py:72">
P2: Queue size and per-user limits are checked outside an atomic transaction, so concurrent enqueues can exceed the limits despite the checks.</violation>

<violation number="2" location="backend/app/db/repositories/execution_queue_repository.py:130">
P2: When queue data is missing, the user count is never decremented, which can permanently inflate per-user limits and block future enqueues.</violation>

<violation number="3" location="backend/app/db/repositories/execution_queue_repository.py:221">
P1: cleanup_stale will delete non-stale entries for priorities 0–9 because the threshold uses BACKGROUND’s priority offset, making the score range include all higher-priority items regardless of timestamp.</violation>
</file>

<file name="backend/workers/run_saga_orchestrator.py">

<violation number="1" location="backend/workers/run_saga_orchestrator.py:41">
P2: Cleanup is not guaranteed if message handling raises or the task is cancelled. Wrap the consumer loop in try/finally so container.close() always runs.</violation>
</file>

<file name="backend/app/services/coordinator/coordinator.py">

<violation number="1" location="backend/app/services/coordinator/coordinator.py:115">
P2: After cancellation, the coordinator doesn’t attempt to schedule the next queued execution. With the scheduling loop removed, this can leave the queue stalled until another event triggers scheduling. Consider calling `_try_schedule_next()` after cancellation cleanup.</violation>
</file>

<file name="backend/tests/e2e/test_k8s_worker_create_pod.py">

<violation number="1" location="backend/tests/e2e/test_k8s_worker_create_pod.py:23">
P2: Using the DI-managed worker ignores `test_settings.K8S_NAMESPACE`; the worker’s `K8sWorkerConfig` namespace comes from the environment, so the test may create resources in a different namespace than the one it asserts/cleans up. This can lead to flaky failures or leaked resources when test settings differ from the env.</violation>
</file>

In `@backend/app/core/providers.py`:
- Around line 131-170: The provider currently hardcodes limits in
RedisRepositoryProvider; update get_execution_queue_repository and
get_resource_repository to read max_queue_size, max_executions_per_user,
total_cpu_cores, and total_memory_mb from the Settings argument instead of
hardcoded literals; add those fields with sensible defaults to the Settings
class (e.g., max_queue_size, max_executions_per_user, total_cpu_cores,
total_memory_mb), then pass settings.max_queue_size and
settings.max_executions_per_user into ExecutionQueueRepository and
settings.total_cpu_cores and settings.total_memory_mb into ResourceRepository
and keep existing initialize() call.

In `@backend/app/db/repositories/execution_queue_repository.py`:
- Around line 218-229: The cleanup_stale method computes a single huge
threshold_score that causes all non-BACKGROUND priorities to be considered
stale; update cleanup_stale to iterate over each QueuePriority and for each
priority compute min_score = priority.value * 1e12 and max_score =
priority.value * 1e12 + (time.time() - self.stale_timeout_seconds) (or
alternatively fetch and check the stored _enqueue_timestamp in the entry hash)
then call zrangebyscore with that per-priority range, decode entries and call
self.remove(execution_id) as before; ensure you still limit batch size (num=100)
and aggregate removed count.
- Around line 71-101: The current enqueue logic performs zcard and hincrby(...,
0) checks outside the pipeline which allows race conditions; change the
implementation in the enqueue routine to perform the queue-size and
per-user-count checks and the zadd + data hset/hincrby/expire atomically by
moving them into a single Redis Lua script (or by using WATCH on QUEUE_KEY and
USER_COUNT_KEY then MULTI/EXEC). Specifically, include the zcard(QUEUE_KEY) and
hincrby(USER_COUNT_KEY, user_id, 0) checks, the zadd(QUEUE_KEY, execution_id,
score), the creation of data_key and its hset/expire, and the
hincrby(USER_COUNT_KEY, user_id, 1) increment inside the atomic operation so the
old pipeline usage (pipe.zadd, pipe.hset, pipe.expire, pipe.hincrby) is executed
only from within the Lua script or guarded MULTI after WATCH; ensure the script
returns appropriate failure/success values (matching the existing
False/None/"Queue is full" or limit-exceeded responses) so the calling code can
handle errors the same way.

In `@backend/app/db/repositories/resource_repository.py`:
- Around line 216-236: The release method has a race where it reads the
allocation then updates the pool separately; change release to perform the read,
increment and delete atomically via a Redis Lua script (similar to allocate()):
create a script that takes KEYS[1]=self.POOL_KEY and KEYS[2]=alloc_key, reads
'cpu','memory','gpu' from the alloc hash (coerce to numbers, default 0), returns
0 if all zero, otherwise calls HINCRBYFLOAT/HINCRBY on the pool fields and DEL
on the alloc key and returns 1; in release() call that script with the two keys
(using self._redis.eval or the client script API), interpret the numeric return
as the boolean result and remove the existing pipeline-based logic and prior
hgetall read. Ensure you reference ALLOC_KEY_PREFIX when building alloc_key and
keep behavior of logging/returning False when script returns 0.
- Around line 117-193: The allocate method currently does a non-atomic
Python-side existence check before running the Lua script; move that existence
check into the Lua script by testing if alloc_key exists (redis.call('EXISTS',
alloc_key)) at the top of lua_script and return a distinct code (e.g., 2) when
it does, keep returning 0 for insufficient resources and 1 for successful
allocation; then update the Python allocate method (the result handling after
self._redis.eval) to treat result == 2 by reading the existing allocation from
alloc_key and returning a ResourceAllocation built from those fields (same
parsing logic used earlier), otherwise proceed as before for success/failure.
- Around line 174-176: alloc_key entries are being set with a 7200s TTL but pool
resource counters (updated in the allocation code around alloc_key and the pool
keys) are never restored if release() isn't called, causing permanent capacity
leaks; add a reconciliation background task (e.g., reclaim_expired_allocations
or reconcile_expired_allocations) that periodically scans redis for keys
matching "alloc_key:*", detects expired or missing allocation metadata and
atomically restores the pool counters (cpu, memory, gpu) by incrementing the
corresponding pool hash fields, using Redis transactions or a Lua script to
avoid races, and mark or remove reclaimed alloc_key entries to make reclamation
idempotent; alternatively add TTLs to pool keys and implement a periodic
reconcile_pools task that verifies sums against active allocations and fixes
discrepancies—update resource_repository functions (alloc_key handling and
release()) to cooperate with the new reconciler.

In `@backend/app/services/coordinator/coordinator.py`:
- Around line 194-204: The current flow dequeues an execution and, on allocation
failure, removes its claim and requeues it via self._state_repo.remove and
self._queue_repo.enqueue, which loses original queue position and can cause
busy-loops; change this by checking resource availability before dequeuing (use
or add a can_allocate or resources_available check in the coordinator) or move
the execution into a "waiting for resources" state instead of re-enqueueing at
the tail, and if you must requeue implement exponential/backoff delay before
calling self._queue_repo.enqueue or mark the item with a waiting flag so dequeue
logic preserves position and avoids tight requeue cycles (update the code paths
around dequeue, allocation, allocation, _state_repo.remove, and
_queue_repo.enqueue accordingly).
- Around line 97-100: handle_execution_requested currently logs exceptions and
records metrics but does not notify downstream systems; update the except block
in handle_execution_requested to call self._publish_scheduling_failed(event,
str(e)) before/after recording the metric so an ExecutionFailedEvent is
published (mirror the behavior in _schedule_execution), keeping the existing
self._logger.error and
self._metrics.record_coordinator_execution_scheduled("error") calls.

In `@backend/app/services/event_bus.py`:
- Around line 74-82: The current try/except around self._producer.send_and_wait
in event_bus (the block catching Exception as e) only logs failures and swallows
the exception; update the error-handling so callers receive failure info —
either re-raise the caught exception after logging or change the method
signature to return a boolean and return False on failure (and True on success);
locate the send_and_wait call and its except Exception as e handler in the
publish/publish_event method, add the chosen behavior (re-raise e after
self._logger.error(...) or return False), and ensure successful paths return
True if using the success indicator approach.

In `@backend/app/services/notification_service.py`:
- Around line 170-174: The truncation code assumes event_data["stdout"] and
["stderr"] are strings but they can be None; update the logic around event_data
(produced by event.model_dump) to guard against None before slicing—e.g.,
replace direct slicing with a safe truncation that uses a default empty string
when the key is None or missing, and apply the same guard for both "stdout" and
"stderr" to prevent TypeError in NotificationService (look for event_data,
"stdout", and "stderr" usage in notification_service.py).

In `@backend/tests/unit/services/pod_monitor/test_monitor.py`:
- Line 15: Replace the incorrect import of ResourceUsageDomain in
test_monitor.py: locate the module that actually defines/exports
ResourceUsageDomain (instead of app.domain.execution.models) and import
ResourceUsageDomain from that module (update the import statement that currently
references ResourceUsageDomain). Ensure the import uses the exact exported
symbol name ResourceUsageDomain so MyPy recognizes it.

In `@backend/tests/unit/services/sse/test_sse_service.py`:
- Around line 13-14: The import for ResourceUsageDomain is incorrect: update the
import line that currently pulls DomainExecution and ResourceUsageDomain from
app.domain.execution so that DomainExecution remains imported from
app.domain.execution and ResourceUsageDomain is imported from
app.domain.events.typed; specifically adjust the import statement referencing
DomainExecution and ResourceUsageDomain in test_sse_service.py so
DomainExecution stays from app.domain.execution and ResourceUsageDomain is
imported from app.domain.events.typed (keeping SSEExecutionStatusDomain and
ShutdownStatus as-is).

In `@backend/workers/run_coordinator.py`:
- Around line 37-48: The async consumer loop in ExecutionCoordinator using
kafka_consumer and handler can raise exceptions which skip cleanup; wrap the
async for msg in kafka_consumer loop in a try/finally so that await
container.close() (and any kafka_consumer commit/close if needed) always runs,
move the "ExecutionCoordinator shutdown complete" log into the finally block,
and ensure resource-specific cleanup for AIOKafkaConsumer and UnifiedConsumer
happens before closing the container (referencing kafka_consumer, handler,
container, AIOKafkaConsumer, UnifiedConsumer).

In `@backend/workers/run_event_replay.py`:
- Around line 32-35: The forever-blocking await asyncio.Event().wait() prevents
container.close() from ever running; replace the anonymous Event with a named
asyncio.Event (e.g., shutdown_event), register signal handlers for
SIGINT/SIGTERM that set shutdown_event (using loop.add_signal_handler or
signal.signal fallback), await shutdown_event.wait() in run_event_replay.py,
then call await container.close() after the wait so the container is closed
during graceful shutdown; reference the existing asyncio.Event().wait() and
container.close() calls when making the change.

In `@backend/workers/run_k8s_worker.py`:
- Around line 36-47: The consumer loop can raise in handler.handle() or
kafka_consumer.commit() and skip cleanup; wrap the async for msg in
kafka_consumer loop in a try/finally so that await container.close() (and any
other shutdown steps) always run. Specifically, enclose the async for ...: await
handler.handle(msg); await kafka_consumer.commit() inside try: ... finally:
await container.close() and keep the logger.info("KubernetesWorker shutdown
complete") inside the finally; apply the same pattern to the other worker
modules (run_result_processor.py, run_saga_orchestrator.py, run_coordinator.py,
run_pod_monitor.py) referencing kafka_consumer, UnifiedConsumer.handle,
kafka_consumer.commit, and container.close where present.

In `@backend/workers/run_result_processor.py`:
- Around line 36-47: Wrap the consumer loop in a try/finally so cleanup always
runs: start by acquiring kafka_consumer (AIOKafkaConsumer) and handler
(UnifiedConsumer) as before, then run the async for msg in kafka_consumer /
await handler.handle(msg) loop inside a try block; in the finally block ensure
you stop/close the kafka_consumer (call kafka_consumer.stop() or equivalent) and
then await container.close(); re-raise cancellation errors if needed so signal
handling still works. This guarantees the consumer and container are closed even
if handler.handle() raises or the task is cancelled.

In `@backend/workers/run_saga_orchestrator.py`:
- Around line 41-47: Add graceful shutdown and per-message error resilience
around the kafka_consumer loop: install asyncio signal handlers for
SIGINT/SIGTERM that set an asyncio.Event (e.g., shutdown_event) and make the
async for msg in kafka_consumer loop respect that event so it stops cleanly;
wrap message processing so each iteration calls await handler.handle(msg) inside
a try/except that logs the exception via logger.error (including exception
details) and continues (optionally still commit offsets on successful processing
via kafka_consumer.commit), and wrap the whole consumer loop in try/finally to
ensure await container.close() and any kafka_consumer stop/close are always
awaited on shutdown. Reference symbols: kafka_consumer, handler.handle,
kafka_consumer.commit, container.close, logger.